/**
 * Copyright 2014 wasu.com
 *
 *
 * Create on 2015年10月21日 下午4:06:55
 */
package xj.toolkit.rpc.core;

import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpResponse;
import io.netty.handler.codec.http.HttpHeaders;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.InputStream;
import java.lang.reflect.InvocationTargetException;
import java.util.Arrays;
import java.util.List;

import javax.management.ServiceNotFoundException;

import org.apache.commons.lang3.ClassUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.caucho.hessian.io.AbstractHessianInput;
import com.caucho.hessian.io.AbstractHessianOutput;
import com.caucho.hessian.io.Hessian2Input;
import com.caucho.hessian.io.HessianFactory;
import com.caucho.hessian.io.HessianInputFactory;
import com.caucho.hessian.io.SerializerFactory;
import com.caucho.services.server.ServiceContext;
import com.esotericsoftware.reflectasm.MethodAccess;
import com.google.common.base.Preconditions;

import xj.toolkit.netty.handler.codec.http.request.HttpRequestExt;
import xj.toolkit.rpc.assemble.RpcServiceDescriptor;
import xj.toolkit.rpc.assemble.RpcServiceManager;

/**
 * @author <a href="mailto:caoxiaojian@wasu.cn">xiaojiancao</a>
 *
 */
@Service("rpcServiceInvocationerHandler")
public class RpcServiceInvocationerHandler {

	private static final Logger logger = LoggerFactory.getLogger(RpcServiceInvocationerHandler.class);

	/**
	 * rpc服务管理者。
	 */
	@Autowired
	private RpcServiceManager rpcServiceManager = null;

	private HessianInputFactory _inputFactory = new HessianInputFactory();
	private HessianFactory _hessianFactory = new HessianFactory();

	private SerializerFactory _serializerFactory = new SerializerFactory();

	public void invoke(RpcServiceDescriptor rpcServiceDescriptor, HttpRequestExt request) throws Exception {
		int contentLength = 0;

		try {
			String value = request.headers().get(HttpHeaders.Names.CONTENT_LENGTH);
			contentLength = Integer.parseInt(value);
		} catch (Exception e) {
			logger.error("Cannot found request length, request[" + request + "].", e);
			throw e;
		}

		byte[] bytes = new byte[contentLength];
		request.content().readBytes(bytes);

		ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
		HessianInputFactory.HeaderType header = _inputFactory.readHeader(bis);

		ByteArrayOutputStream bos = new ByteArrayOutputStream();

		AbstractHessianInput in = null;
		AbstractHessianOutput out = null;

		switch (header) {
		case CALL_1_REPLY_1:
			in = _hessianFactory.createHessianInput(bis);
			out = _hessianFactory.createHessianOutput(bos);
			break;

		case CALL_1_REPLY_2:
			in = _hessianFactory.createHessianInput(bis);
			out = _hessianFactory.createHessian2Output(bos);
			break;

		case HESSIAN_2:
			in = _hessianFactory.createHessian2Input(bis);
			in.readCall();
			out = _hessianFactory.createHessian2Output(bos);
			break;

		default:
			throw new IllegalStateException(header + " is an unknown Hessian call");
		}

		in.setSerializerFactory(_serializerFactory);
		out.setSerializerFactory(_serializerFactory);

		invoke(rpcServiceDescriptor, in, out);

		FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);

		byte[] content = bos.toByteArray();
		response.content().writeBytes(content);
		response.headers().set(HttpHeaders.Names.CONTENT_LENGTH, content.length);
		response.headers().set(HttpHeaders.Names.CONTENT_TYPE, "applicaiton/octet-stream");

		SendHepler.sendResponse(request, response);

	}

	protected void invoke(RpcServiceDescriptor rpcServiceDescriptor, AbstractHessianInput in, AbstractHessianOutput out)
			throws Exception {
		ServiceContext context = ServiceContext.getContext();

		// backward compatibility for some frameworks that don't read
		// the call type first
		in.skipOptionalCall();

		// Hessian 1.0 backward compatibility
		String header;
		while ((header = in.readHeader()) != null) {
			Object value = in.readObject();

			context.addHeader(header, value);
		}

		String methodName = in.readMethod();
		int argLength = in.readMethodArgLength();
		Object[] argObjs = new Object[argLength];
		Class<?>[] argsTypes = new Class[argObjs.length];
		for (int i = 0; i < argLength; i++) {
			argObjs[i] = in.readObject();
			argsTypes[i] = argObjs[i].getClass();
		}

		Object result = null;

		try {
			// handle request
			result = handleRpcCall(rpcServiceDescriptor.getServiceName(), methodName, argsTypes, argObjs);
		} catch (Exception e) {
			Throwable e1 = e;
			if (e1 instanceof InvocationTargetException)
				e1 = ((InvocationTargetException) e).getTargetException();

			logger.debug(this + " " + e1.toString(), e1);
			result = e;
			out.writeFault("ServiceException", e1.getMessage(), e1);
			out.close();
			return;
		}

		// The complete call needs to be after the invoke to handle a
		// trailing InputStream
		in.completeCall();

		out.writeReply(result);

		out.close();
	}

	protected Hessian2Input createHessian2Input(InputStream is) {
		return new Hessian2Input(is);
	}

	/**
	 * 调用rpc服务的实现类。
	 * 
	 * @param serviceName
	 *            服务名称。
	 * @param methodName
	 *            调用方法名称。
	 * @param args
	 *            参数值。
	 * @return 调用结果。
	 */
	protected Object handleRpcCall(String serviceName, String methodName, Class<?>[] argsTypes, Object[] args)
			throws ServiceNotFoundException {
		Preconditions.checkArgument(StringUtils.isNoneBlank(serviceName));
		Preconditions.checkArgument(StringUtils.isNoneBlank(methodName));

		RpcServiceDescriptor descriptor = rpcServiceManager.getRpcServiceByName(serviceName);
		if (descriptor == null) {
			throw new ServiceNotFoundException("Service[" + serviceName + "] not found.");
		}

		MethodAccess methodAccess = descriptor.getMethodAccess();
//		int methodIndex = methodAccess.getIndex(methodName, argsTypes);
		int methodIndex = -1;
		for (int i = 0; i < methodAccess.getMethodNames().length; i++) {
			if (!methodAccess.getMethodNames()[i].equals(methodName)) {
				continue;
			}
			
			if (argsTypes == null || argsTypes.length == 0) {
				methodIndex = i;
				break;
			}
			
			if (compareArgs(methodAccess.getParameterTypes()[i], argsTypes)) {
				methodIndex = i;
				break;
			}
		}
		
		if (methodIndex == -1) {
			throw new IllegalArgumentException("Unable to find non-private method: " + methodName + " " + Arrays.toString(argsTypes));
		}
		
		return methodAccess.invoke(descriptor.getBeanInstance(), methodIndex, args);
	}
	
	/**
	 * 检查参数类型，参数类型要么相同，要么入参的参数类型的接口类在方法参数里面。
	 * 
	 * @param mehtodArgsTypes 从method拿出的方法参数类型列表。
	 * @param argsTypes 需要比对的方法参数类型列表
	 * @return
	 */
	protected boolean compareArgs(Class<?>[] mehtodArgsTypes, Class<?>[] argsTypes) {
		if (mehtodArgsTypes.length != argsTypes.length) {
			return false;
		}
		
		for (int i = 0; i < mehtodArgsTypes.length; i++) {
			if (mehtodArgsTypes[i] == argsTypes[i]) {
				continue;
			}
			
			List<Class<?>> allInterfaces = ClassUtils.getAllInterfaces(argsTypes[i]);
			if (allInterfaces.contains(mehtodArgsTypes[i])) {
				continue;
			}
			
			return false;
		}
		
		return true;
	}

}
